package com.test.ant.payment.handler;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.test.ant.log.FileWorkThread;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.charset.Charset;
import java.time.Duration;
import java.time.Instant;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.concurrent.*;

public class WrongTypeCountService {

    /**
     * 默认最大线程数量
     * <p>
     * ((线程等待时间+线程CPU时间 ) / 线程CPU时间 ) * CPU数目
     */

    public static String CURRENT_WORK_DIR = "/logs/";
    private static String VARIABLE = "Read";

    public static void main(String[] args) throws IOException, InterruptedException {
        Instant start = Instant.now();
        Map<String, Integer> atomicReference = SplitLogFile.getCountMaps(CURRENT_WORK_DIR + "aa.log");
        System.out.println("处理文件花费：" + Duration.between(start, Instant.now()).toMillis() / 1000.0 + "秒");
        System.out.println(VARIABLE + "错误类型个数 ：" + atomicReference.get(VARIABLE));
    }


}

class SplitLogFile {
    private File file = null;
    private Vector<CountWrongTypeThread> listCountWrongTypeThreads = null;
    private long splitSize;
    private long currentPos;

    public SplitLogFile(File file, long splitSize) throws IOException, InterruptedException {
        this.file = file;
        this.splitSize = splitSize;
        this.currentPos = 0;
        this.listCountWrongTypeThreads = new Vector<CountWrongTypeThread>();
    }

    private static final int DEFAULT_MAX_THREAD = ((1 + 2) * 2) * Runtime.getRuntime().availableProcessors();
    private static final ExecutorService executor = new ThreadPoolExecutor(
            DEFAULT_MAX_THREAD,
            DEFAULT_MAX_THREAD,
            0L,
            TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(1024),
            new ThreadFactoryBuilder().setDaemon(true).setNameFormat("RealTime PaymentTypeEnableFilter-%d").build(),
            new ThreadPoolExecutor.CallerRunsPolicy());

    public static Map<String, Integer> getCountMaps(String fileName) throws IOException, InterruptedException {
        File file = new File(fileName);
        SplitLogFile slf = new SplitLogFile(file, 1024 * 1); // 文件，文件分割大小
        return slf.doCount();
    }

    public Map<String, Integer> doCount() throws IOException {
        // 获取文件分割的线程集合
        List<CountWrongTypeThread> threadList = doSplitFile();
        // 设置线程的计数器
        CountDownLatch countDownLatch = new CountDownLatch(threadList.size());
        threadList.forEach(th -> th.setCountDownLatch(countDownLatch));

        // 线程池执行
        Instant start = Instant.now();
        threadList.forEach(th -> executor.execute(th));

        // 主线程 await
        Instant end = null;
        try {
            countDownLatch.await();
            end = Instant.now();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 线程池关闭
        executor.shutdown();
        // 执行时间计算和模拟查询一个数值
        System.out.println("Time = " + Duration.between(start, end).toMillis() / 1000.0 + "秒");
        Map<String, Integer> map = new HashMap<>();
        for (CountWrongTypeThread countWrongTypeThread : threadList) {
            map = countWrongTypeThread.getResultMap();
        }
        return map;
    }

    public List<CountWrongTypeThread> doSplitFile() throws IOException {
        while (currentPos < this.file.length()) {
            for (int num = 0; num < DEFAULT_MAX_THREAD; num++) {
                if (currentPos < file.length()) {
                    CountWrongTypeThread countWrongTypeThread = null;
                    if (currentPos + splitSize < file.length()) {
                        RandomAccessFile raf = new RandomAccessFile(file, "r");
                        raf.seek(currentPos + splitSize);
                        int offset = 0;
                        while (true) {
                            char ch = (char) raf.read();
                            //末尾跳出
                            if (-1 == ch) {
                                break;
                            }
                            //是否是字母和'，都不是跳出（防止单词被截断）
                            if (false == Character.isLetter(ch) && '\'' != ch) {
                                break;
                            }
                            offset++;
                        }
                        countWrongTypeThread = new CountWrongTypeThread(file, currentPos, splitSize + offset);
                        currentPos += splitSize + offset;
                        raf.close();
                    } else {
                        countWrongTypeThread = new CountWrongTypeThread(file, currentPos, file.length() - currentPos);
                        currentPos = file.length();
                    }
                    Thread thread = new Thread((Runnable) countWrongTypeThread);
                    thread.start();
                    listCountWrongTypeThreads.add(countWrongTypeThread);
                }
            }
        }
        return listCountWrongTypeThreads;
    }
}

class CountWrongTypeThread implements Runnable {
    private FileChannel fileChannel = null;
    private FileLock lock = null;
    private MappedByteBuffer mbBuf = null;
    private Map<String, Integer> hashMap = null;

    /**
     * 计数器
     */
    private CountDownLatch countDownLatch;

    /**
     * 同步计数器
     *
     * @param countDownLatch 计数器
     */
    public void setCountDownLatch(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
    }

    public CountWrongTypeThread(File file, long start, long size) { //文件，起始位置，映射文件大小
        try {
            fileChannel = new RandomAccessFile(file, "rw").getChannel();
            lock = fileChannel.lock(start, size, false);
            // 如果文件过大需要切割成多个片段
            mbBuf = fileChannel.map(FileChannel.MapMode.READ_ONLY, start, size);
            hashMap = new HashMap<String, Integer>();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void run() {
        String str = Charset.forName("UTF-8").decode(mbBuf).toString();
        String[] strArray = str.split(System.getProperty("line.separator"));
        for (int i = 0; i < strArray.length; i++) {
            if (strArray[i].equals("")) {
                continue;
            }
            int endIndex = strArray[i].indexOf(" ", 1);
            String key = strArray[i].substring(1, endIndex);
            if (hashMap.get(key) == null) {
                hashMap.put(key, 1);
            } else {
                hashMap.put(key, hashMap.get(key) + 1);
            }
        }
        try {
            lock.release();
            fileChannel.close();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            countDownLatch.countDown();
        }
    }

    //获取当前线程的执行结果
    public Map<String, Integer> getResultMap() {
        return hashMap;
    }
}